package hust.rpc.invoker;

import hust.rpc.lb.LoadBalance;
import hust.rpc.pojo.*;
import hust.rpc.utils.CommonUtils;
import hust.rpc.utils.SignatureUtils;
import hust.rpc.utils.SingletonUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class RemoteInvoker {

    protected final Map<String, CompletableFuture<RpcResponse>> responses = new ConcurrentHashMap<>();

    protected final Map<String, Set<Channel>> serviceToChannels = new ConcurrentHashMap<>();
    protected final Map<Channel, List<String>> channelToServices = new ConcurrentHashMap<>();

    public void addProviders(Map<Host, List<String>> providerInfo) {
        providerInfo.forEach((host, services) -> addProvider(Channel.wrap(host), services));
    }

    public void addProvider(Channel channel, List<String> services) {
        // 先清除原先的映射
        channelToServices.remove(channel);
        // 更新映射
        channelToServices.put(channel, services);
        // 更新反向映射
        for (String service : services) {
            Set<Channel> channels = serviceToChannels.get(service);
            if (channels != null) {
                synchronized (channels) {
                    channels.add(channel);
                }
            } else {
                synchronized (channelToServices) {
                    channels = serviceToChannels.get(service);
                    if (channels != null) {
                        synchronized (channels) {
                            channels.add(channel);
                        }
                        continue;
                    }
                    channels = new HashSet<>();
                    channels.add(channel);
                    serviceToChannels.put(service, channels);
                }
            }
        }
    }

    public void delProvider(Channel channel) {
        // 删除映射
        List<String> services = channelToServices.remove(channel);
        for (String service : services) {
            Set<Channel> channels = serviceToChannels.get(service);
            synchronized (channels) {
                channels.remove(channel);
            }
        }
    }

    /*public Map<String, List<Host>> getProviders(List<String> services) {
        Map<String, List<Host>> result = new HashMap<>();
        for (String service : services) {
            Set<Channel> channels = serviceToChannels.get(service);
            if (channels.isEmpty()) continue;
            List<Host> hosts = channels
                    .stream()
                    .map(Channel::getHost)
                    .collect(Collectors.toList());
            result.put(service, hosts);
        }
        return result;
    }

    public Map<String, List<Host>> getProviders(List<String> services, Map<Channel, Channel> channelMap) {
        Map<String, List<Host>> result = new HashMap<>();
        for (String service : services) {
            Set<Channel> channels = serviceToChannels.get(service);
            if (channels.isEmpty()) continue;
            List<Host> hosts = channels
                    .stream()
                    .map(it -> channelMap.get(it).getHost())
                    .collect(Collectors.toList());
            result.put(service, hosts);
        }
        return result;
    }*/

    public void complete(Host host, RpcResponse response) {
        String key = host.getIp() + ':' + host.getPort() + ':' + response.getId();
        CompletableFuture<RpcResponse> future = responses.remove(key);
        if (future != null && !future.isCancelled()) {
            future.complete(response);
        } else {
            // TODO 统一异常
            throw new RuntimeException("complete调用错误!");
        }
    }

    public CompletableFuture<RpcResponse> sendRequest(Message message, Channel channel) {
        RpcRequest request = (RpcRequest) message.getData();
        log.info("发送信息{}给通道{}", message, channel);
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        // 如果是消费者,可以事先没有Channel,只有Host,此时需要调用ChannelProvider进行调用
        if (channel.getChannel() == null) {
            handleNoChannel(channel);
        }
        if (!channel.getChannel().isActive()) {
            throw new RuntimeException("Channel is inactive!");
        }
        channel.getChannel().writeAndFlush(message);
        responses.put(CommonUtils.getFutureKey(channel.getHost(), request.getId()), future);
        return future;
    }

    protected void handleNoChannel(Channel channel) {
        delProvider(channel);
        throw new RuntimeException("Channel无法提供服务!");
    }

    public CompletableFuture<RpcResponse> sendRequest(Message message) {
        RpcRequest request = (RpcRequest) message.getData();
        String service = SignatureUtils.signatureToService(request.getSignature());
        if (service == null) {
            throw new RuntimeException("找不到目标服务!");
        }
        LoadBalance lb = SingletonUtils.getInstance(LoadBalance.class);
        Set<Channel> channels = serviceToChannels.get(service);
        Channel channel = lb.next(new ArrayList<>(channels),request);
        return sendRequest(message, channel);
    }
}
